mod signal;

use std::{
    cell::{LazyCell, RefCell},
    error,
    ops::Deref,
};

use cs2s::{
    cache::CacheTableVal,
    command::{
        Command, CommandKind, CommandQueueServerHandle, CommandQueueStatus, TerminateStatus,
    },
    error::Error,
    route::HandleRouter,
    utils::map::Key,
};
use signal_hook::{
    consts::{SIGINT, SIGTERM},
    iterator::Signals,
};

use crate::{args::ServerStartCommand, ARGS};

thread_local! {
    /// Global Cache tables.
    static CACHE_TABLES: LazyCell<RefCell<HandleRouter>> = LazyCell::new(|| {
        RefCell::new(HandleRouter::new(&ARGS.name))
    });
}

/// Start the server and listen for commands.
pub fn start(command: &ServerStartCommand) -> Result<(), Box<dyn error::Error>> {
    env_logger::Builder::new()
        .format_target(false)
        .filter_level(command.log_level)
        .init();

    info!("starting server: {}", ARGS.name);

    let mut queue = CommandQueueServerHandle::new(&ARGS.name)?;

    signal::handle(&ARGS.name, Signals::new([SIGINT, SIGTERM])?);

    let status = loop {
        match queue.listen((handle, |status: CommandQueueStatus| {
            report(status, command.congestion_threshold)
        })) {
            Ok(status) => break status,
            Err(err) => match err.downcast_ref::<Error>() {
                Some(Error::InvalidClient { pid }) => warn!("client {} is invalid", pid),
                _ => return Err(err),
            },
        }
    };

    for client in queue.connected_clients() {
        warn!("client {} has not disconnected yet", client);
    }

    match status {
        TerminateStatus::Success => {
            info!("terminated successfully");
            Ok(())
        }
        TerminateStatus::Failure => Err("terminated with failure".into()),
    }
}

/// Handle the command.
fn handle(command: &Command, payload: Vec<&[u8]>) -> Result<(), Error> {
    if !payload.is_empty() {
        info!(
            "received {} with payloads ({})",
            command,
            payload
                .iter()
                .map(|payload| format!("{} B", payload.len()))
                .collect::<Vec<_>>()
                .join(", ")
        );
    } else {
        info!("received {}", command);
    }

    if let Err(err) = dispatch(command, payload) {
        error!("{}", err);
    }

    Ok(())
}

/// Report the status of the command queue.
fn report(status: CommandQueueStatus, congestion_threshold: f64) {
    let capacity = status.capacity();
    let current = status.current();

    let factor = current as f64 / capacity as f64;
    if factor >= congestion_threshold {
        warn!(
            "congestion detected: {}/{} ({:.3})",
            current, capacity, factor,
        );
    } else {
        debug!(
            "congestion status: {}/{} ({:.3})",
            current, capacity, factor,
        );
    }
}

/// Dispatch the command.
fn dispatch(command: &Command, payload: Vec<&[u8]>) -> Result<(), Box<dyn error::Error>> {
    match command.kind() {
        CommandKind::PathsAttach { .. } => {
            CACHE_TABLES.with(|cache_tables| {
                cache_tables.deref().borrow_mut().create(
                    payload
                        .into_iter()
                        .map(|path| String::from_utf8_lossy(path).to_string()),
                )
            })?;
        }
        CommandKind::Synchronize { cache_key, .. } => {
            CACHE_TABLES.with(|cache_tables| {
                cache_tables.deref().borrow_mut().synchronize(
                    String::from_utf8_lossy(payload[0]),
                    &cache_key.as_ref(),
                    &CacheTableVal {
                        host_meta: payload[1],
                        host_code: payload[2],
                    },
                )
            })?;
        }
        _ => {}
    }

    Ok(())
}
